package com.kuizii.discovery.net;

import com.kuizii.common.LiteServiceRegistration;
import com.kuizii.common.net.EventType;
import com.kuizii.common.net.LiteRegisterRequest;
import com.kuizii.discovery.client.DiscoveryProperties;
import com.kuizii.discovery.exception.DiscoveryIOException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class NettyIOManager implements IOManager {

    ContentSerialize serialize;
    private Bootstrap bootstrap;
    private Channel channel;
    DiscoveryProperties discoveryProperties;

    public NettyIOManager() {
        new NettyIOManager(new JsonSerialize());
    }

    public NettyIOManager(ContentSerialize serialize) {
        this.serialize = serialize;
    }

    @Override
    public void init(DiscoveryProperties discoveryProperties) {
        bootstrap = new Bootstrap();
        this.discoveryProperties = discoveryProperties;

        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        try {
            bootstrap.group(bossGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel nioSocketChannel) throws Exception {
                    ChannelPipeline pipeline = nioSocketChannel.pipeline();
                    channel = pipeline.channel();

                    //  pipeline.addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS));
//                    pipeline.addLast(new HeartBeatClientHandler());
//                    pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
//                    pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast(new ObjectEncoder());
                    pipeline.addLast(new ObjectDecoder(ClassResolvers.weakCachingResolver(this.getClass().getClassLoader())));
//                    pipeline.addLast(new DefaultMessageEncoder());
//                    pipeline.addLast(new DefaultMessageDecoder());
                    pipeline.addLast(new DefaultMessageSenderHandler());

//                    nioSocketChannel.writeAndFlush("开始连接");
                }
            });

            String serverAddr = discoveryProperties.getServerAddr();
            if (StringUtils.isEmpty(serverAddr) || !serverAddr.contains(":")) {
                throw new DiscoveryIOException("Emm..请检查远程地址 !");
            }
            String[] addrs = serverAddr.split(":");
            int port = 9527;
            if (addrs.length >= 2) {
                port = Integer.valueOf(addrs[1]);
            }
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.remoteAddress(addrs[0], port);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
//            bossGroup.shutdownGracefully();
        }
        doConnect();
    }

    protected void doConnect() {

        if (channel != null && channel.isActive()) {
            return;
        }
        try {
            ChannelFuture f = bootstrap.connect().sync();
            f.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        channel = channelFuture.channel();
                        System.out.println("连接成功!");

                    } else {
                        System.out.println("连接失败,稍后将重新连接");
                        f.channel().eventLoop().schedule(new Runnable() {
                            @Override
                            public void run() {
                                doConnect();
                            }
                        }, 10, TimeUnit.SECONDS);
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
//            bossGroup.shutdownGracefully();
        }
    }


    @Override
    public void registerService(LiteServiceRegistration instance) {

        doConnect();
        System.out.println("注册消息");
        LiteRegisterRequest registerRequest = new LiteRegisterRequest();
        registerRequest.setEventType(EventType.REGISTER);
        registerRequest.setRegistration(instance);
        ChannelFuture channelFuture = channel.writeAndFlush(registerRequest);
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                refreshService(instance.getServiceId(), instance);
            }
        });

    }

    @Override
    public void deregisterService(String serviceName, LiteServiceRegistration instance) {
        doConnect();
        System.out.println("卸载消息");

        LiteRegisterRequest registerRequest = new LiteRegisterRequest();
        registerRequest.setEventType(EventType.DEREGISTER);
        registerRequest.setRegistration(instance);
        channel.writeAndFlush(registerRequest);
        try {
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void refreshService(String serviceName, LiteServiceRegistration instance) {

        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
        scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                doConnect();
                System.out.println("刷新注册消息");

                LiteRegisterRequest registerRequest = new LiteRegisterRequest();
                registerRequest.setEventType(EventType.HEALTH_CHECK);
                registerRequest.setRegistration(instance);
                channel.writeAndFlush(registerRequest);
            }
        }, 5, 10, TimeUnit.SECONDS);
    }

    @Override
    public void destroy() {
        if (channel != null) {
            try {
                System.out.println("关闭连接");
                channel.close().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    @Override
    public List getServerList() {
        return null;
    }
}
